package com.practise.rocketmqdemo.listen;

import com.practise.rocketmqdemo.constant.MQConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.stereotype.Component;

/**
 ** topic需要和生产者的topic一致，consumerGroup属性是必须指定的，内容可以随意
 ** selectorExpression的意思指的就是tag，默认为“*”，不设置的话会监听所有消息
 */
@Slf4j
@Component
@RocketMQTransactionListener()
public class TopicTransactionReceiver implements RocketMQLocalTransactionListener {


    /**
     * executeLocalTransaction方法，当我们处理完业务后，可以根据业务处理情况，返回事务执行状态，
     * 有bollback, commit or unknown三种，分别是回滚事务，提交事务和未知；根据事务消息执行流程，
     * 如果返回bollback，则直接丢弃消息；如果是返回commit，则消费消息；如果是unknow，
     * 则继续等待，然后调用checkLocalTransaction方法，最多重试15次，超过了默认丢弃此消息；
     * @param msg
     * @param arg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        MessageHeaders headers = msg.getHeaders();
        String transactionId = (String)headers.get(RocketMQHeaders.TRANSACTION_ID);
        String msgId = (String)headers.get("msg_id");
        try {
            // 执行业务逻辑
            log.info("=======>接受到事务消息 msg = {}",msg.getPayload());
            //本地事物成功，执行commit
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            log.error("本地事物执行异常，e={}",e);
            //本地事物失败，执行rollback
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    /**
     * 是当MQ Server未得到MQ发送方应答，或者超时的情况，或者应答是unknown的情况，调用此方法进行检查确认
     * @param msg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        MessageHeaders headers = msg.getHeaders();
        String transactionId = (String)headers.get(RocketMQHeaders.TRANSACTION_ID);
         // 检查消费者方的业务日志是否已经成功消费
        boolean isok=true;

        if(!isok){
            log.error("如果本地事物日志没有记录，transactionId={}",transactionId);
            //本地事物失败，执行rollback
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        //如果本地事物日志有记录，执行commit
        return RocketMQLocalTransactionState.COMMIT;
    }
}